package com.sunday.common.mq.rabbit.study.base.e17_Streams;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.sunday.common.mq.rabbit.study.base.ConnectionUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * https://www.rabbitmq.com/streams.html
 * 流
 * 流是一种新的持久和复制的数据结构，它建模 具有非破坏性使用者语义的仅追加日志。
 * 它们可以通过 RabbitMQ 客户端库（就好像它是队列一样）或通过专用的二进制协议插件和相关客户端使用。建议使用后一个选项，因为它 提供对所有特定于流的功能的访问，并提供最佳吞吐量（性能）。
 * 本页介绍流的概念、流的用法以及它们的管理和维护操作。
 * 请访问流插件页面，了解有关使用二进制 RabbitMQ 流协议的流以及功能矩阵的流核心和流插件比较页面的更多信息。
 * <p>
 * 使用案例
 * 开发流最初涵盖 4 个消息传递用例，这些用例 现有的队列类型要么不能提供，要么提供缺点：
 * <p>
 * 大型扇出
 * 当前想要将同一消息传递给多个订阅者用户时 必须为每个使用者绑定一个专用队列。如果使用者数量为 大 这会变得可能效率低下，尤其是在需要持久性时 和/或复制。流将允许任意数量的消费者消费 以非破坏性的方式从同一队列中获取相同的消息，从而消除了需求 以绑定多个队列。流使用者也将能够从副本中读取 允许读取负载分布在群集中。
 * <p>
 * 回放/时间旅行
 * 由于所有当前的 RabbitMQ 队列类型都具有破坏性消费行为，即消息 当消费者完成它们时，从队列中删除，它不是 可以重新读取已使用的消息。流将允许 使用者在日志中的任何点附加并从那里读取。
 * <p>
 * 吞吐量性能
 * 没有持久队列类型能够提供可与之竞争的吞吐量 任何现有的基于日志的邮件系统。流已设计 以性能为主要目标。
 * <p>
 * 大型日志
 * 大多数 RabbitMQ 队列被设计为收敛到空状态，并且 经过优化，当 给定队列。流旨在将大量数据存储在 以最小的内存开销高效的方式。
 * <p>
 * =================================================
 * <p>
 * 其他
 * 以下操作可以以类似于经典队列和仲裁队列的方式使用 但有些具有一些特定于队列的行为。
 * <p>
 * 声明
 * 队列删除
 * 发布商确认
 * 消费（订阅）：消费需要 QoS 要设置的预取。acks作为一种信用机制来推进当前的 消费者的偏移量。
 * 为使用者设置 QoS 预取
 * 使用者确认（牢记 QoS 预取限制）
 * 取消消费者
 *
 * =================================================================
 *
 * 1.Stream可以允许任意数目的消费者消费相同队列中的相同信息，用一种不删除的方式。这样就避免了需要普通队列QUEUE时代需要绑定很多队列的情况。
 * 2.实现了允许重读已经消费过的消息。Stream允许消费者从日志中的任意时间点开始读
 * 3. Stream的吞吐量表现更好
 * 4.之前RabbitMQ的队列在队列中有很多数据时性能不好，Stream被设计为允许用一种高效的，占用更少内存的方式存储大量的数据。
 */
@Slf4j
public class Streams {

    /**
     * 声明
     * 若要声明流，请将 x-queue-type 的队列参数设置为 stream（默认值为 classic）。
     * 此参数必须由客户端提供 在申报时;它不能使用策略设置或更改。
     * 这是因为策略定义或适用的策略可以动态更改，但 队列类型不能。必须在声明时指定。
     * <p>
     * 声明将 x-queue-type 参数设置为 stream 的队列将创建一个流 在每个配置的 RabbitMQ 节点上都有一个副本。
     * 流是仲裁系统 因此，强烈建议使用不均匀的群集大小。
     */
    @Test
    @SneakyThrows
    public void declaring() {

        String QUEUE_NAME = "stream_queue";

        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel channel = connection.createChannel();

            channel.queueDelete(QUEUE_NAME);
            channel.queueDeclare(QUEUE_NAME, true, false, false, Collections.singletonMap("x-queue-type", "stream"));

        }

    }

    /**
     * /**
     * 流支持 3 个额外的队列参数，最好使用策略进行配置
     * <p>
     * x-max-length-bytes
     * 设置流的最大大小（以字节为单位）。请参阅保留。默认值：未设置。
     * <p>
     * x-max-age
     * 设置流的最长期限。请参阅保留。默认值：未设置。
     * <p>
     * x-stream-max-segment-size-bytes
     * 单位：字节。流在磁盘上划分为固定大小的段文件。 此设置控制这些的大小。 默认值：（500000000 字节）。
     */
    @Test
    @SneakyThrows
    public void clientOperationsProducer() {

        String QUEUE_NAME = "stream_queue";

        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel channel = connection.createChannel();

            channel.queueDelete(QUEUE_NAME);

            Map<String, Object> arguments = new HashMap<>();
            arguments.put("x-queue-type", "stream");
            arguments.put("x-max-length-bytes", 20_000_000_000L); // maximum stream size: 20 GB
            arguments.put("x-stream-max-segment-size-bytes", 100_000_000); // size of segment files: 100 MB

            channel.queueDeclare(QUEUE_NAME, true, false, false, arguments);

            for (int i = 0; i < 500; i++) {
                AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                        .priority(i)        //优先级
                        .build();
                channel.basicPublish("", QUEUE_NAME, props, String.valueOf(i).getBytes(StandardCharsets.UTF_8));
            }

        }

    }

    /**
     * 客户端操作
     * 消费
     * 由于流永远不会删除任何消息，因此任何消费者都可以开始读取/消费 从日志中的任何点。这由 x-stream-offset 使用者参数控制。
     * 如果未指定，则使用者将从写入的下一个偏移量开始读取 到使用者启动后的日志。支持以下值：
     * <p>
     * first - 从日志中的第一个可用消息开始
     * last - 从最后写入的消息“块”（一个块 是流中使用的储运单元，简单来说就是一批 由几到几千条消息组成的消息，具体取决于入口）
     * next - 与不指定任何偏移量相同
     * Offset - 指定要附加到日志的确切偏移量的数值。 如果此偏移量不存在，它将分别固定到日志的开始或结束。
     * Timestamp - 指定要附加到日志的时间点的时间戳值。 它将钳制到最接近的偏移量，如果时间戳超出流的范围，它将分别夹紧日志的开始或结束。 对于 AMQP 0.9.1，使用的时间戳是精度为 00 秒的 POSIX 时间，即自 00：00：1970 UTC、01-01-<> 以来的秒数。 请注意，使用者可以接收在指定时间戳之前发布的消息。
     * Interval - 一个字符串值，指定相对于当前时间的时间间隔来附加日志。使用与 x-max-age 相同的规格（请参阅保留)
     */
    @Test
    @SneakyThrows
    public void clientOperationsConsumer() {

        String QUEUE_NAME = "stream_queue";

        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel channel = connection.createChannel();

            channel.basicQos(5);

            channel.basicConsume(QUEUE_NAME, false,
                    // 第一个偏移量规范
                    Collections.singletonMap("x-stream-offset", "first"),
                    // 如何指定要从中消耗的特定偏移量
//                    Collections.singletonMap("x-stream-offset", 300),
                    // 如何指定要使用的特定时间戳
//                    Collections.singletonMap("x-stream-offset", new Date(System.currentTimeMillis() - 60 * 60 * 1_000)),
                    // 最后一个偏移量规范
//                    Collections.singletonMap("x-stream-offset", "last"),
                    (consumerTag, delivery) -> {
                        String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                        log.info(" [x] Received '{}'", message);
                        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                    },
                    consumerTag -> {
                    });

            TimeUnit.SECONDS.sleep(60);

        }

    }


    /**
     * https://blog.rabbitmq.com/posts/2022/07/rabbitmq-3-11-feature-preview-single-active-consumer-for-streams/
     * https://github.com/acogoluegnes/rabbitmq-stream-single-active-consumer
     * 单一活跃消费者
     * 流的单个活动使用者是 RabbitMQ 3.11 及更高版本中提供的一项功能。 它在流上提供独占消耗和消耗连续性。 当共享同一流和名称的多个使用者实例启用单个活动使用者时，一次只有一个实例处于活动状态，因此将接收消息。 其他实例将处于空闲状态。
     * <p>
     * 单个活动使用者功能提供 2 个好处：
     * <p>
     * 消息按顺序处理：一次只有一个消费者。
     * 保持消费连续性：如果活跃的消费者停止或崩溃，组中的消费者将接管。
     *
     * 超级流
     * 超级流是一种通过将大型流分区为较小流来横向扩展的方法。 它们与单个活动使用者集成，以在分区内保留消息顺序。 超级流在 RabbitMQ 3.11 及更高版本中可用。
     * 超级流是由单个常规流组成的逻辑流。 这是一种使用 RabbitMQ 流扩展发布和使用的方法：将大型逻辑流划分为分区流，拆分多个集群节点上的存储和流量。
     * 超级流仍然是一个逻辑实体：由于客户端库的智能性，应用程序将其视为一个“大”流。 超级流的拓扑基于 AMQP 0.9.1 模型，即它们之间的交换、队列和绑定。
     * 可以使用任何 AMQP 0.9.1 库或管理插件创建超级流的拓扑，它需要创建一个直接交换，即“分区”流，并将它们绑定在一起。 不过，使用 rabbitmq-streams add_super_stream命令可能更容易。
     */


}
